package jmine.tec.utils.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Um {@link AbstractExecutorService} que nao tem limite, com apenas uma thread que eh ligada somente quando ha trabalho a ser feito e desligada
 * caso contrario. Nao ha garantias de performance, mas garante que nao mantem uma thread viva se nao for necessario.
 * 
 * @author takeshi
 */
public class SimpleThreadPoolExecutor extends AbstractExecutorService {

    private volatile WorkThread thread;

    private static final int TERMINATING = 1;

    private static final int SHUTDOWN = 2;

    private final AtomicInteger state = new AtomicInteger();

    private final Object lock = new Object[]{};

    private final Queue<Runnable> queue = new LinkedBlockingQueue<Runnable>();

    /**
     * Devolve a proxima tarefa a ser executada ou null se nao ha nada a ser executado
     * 
     * @return Runnable
     */
    protected Runnable next() {
        if (queue.isEmpty()) {
            return null;
        }
        return queue.poll();
    }

    /**
     * {@inheritDoc}
     */
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        Thread t = thread;
        if (t == null) {
            return true;
        }
        t.join(unit.toMillis(timeout));
        synchronized(lock) {
            return thread == null;
        }
    }

    /**
     * {@inheritDoc}
     */
    public boolean isShutdown() {
        return state.get() == TERMINATING;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isTerminated() {
        return state.get() == SHUTDOWN;
    }

    /**
     * {@inheritDoc}
     */
    public void shutdown() {
        state.set(TERMINATING);
    }

    /**
     * {@inheritDoc}
     */
    public List<Runnable> shutdownNow() {
        state.set(SHUTDOWN);
        Thread t = thread;
        if (t != null) {
            t.interrupt();
        }
        ArrayList<Runnable> local = new ArrayList<Runnable>(this.queue);
        this.queue.clear();
        return local;
    }

    /**
     * {@inheritDoc}
     */
    public void execute(Runnable command) {
        if (state.get() == 0) {
            queue.add(command);
            if (thread == null) {
                startThread();
            }
        } else {
            throw new RejectedExecutionException("SimpleThreadPool is terminating");
        }
    }

    /**
     * Inicializa a thread
     */
    private void startThread() {
        synchronized (lock) {
            if (thread == null || thread.stopping) {
                thread = new WorkThread();
                thread.start();
            }
        }
    }

    /**
     * Para a thread
     */
    private void stopThread() {
        synchronized (lock) {
            if (Thread.currentThread() == thread) {
                thread = null;
            }
            state.compareAndSet(TERMINATING, SHUTDOWN);
        }
    }

    /**
     * @author takeshi
     */
    private final class WorkThread extends Thread {

        private volatile boolean stopping;
        
        /**
         * {@inheritDoc}
         */
        public void run() {
            if (state.get() != SHUTDOWN) {
                Runnable task = next();
                stopping = task == null || Thread.interrupted();
                while (!stopping) {
                    task.run();
                    task = next();
                    stopping = task == null || Thread.interrupted();
                }
            }
            stopThread();
        }
    }

}
